package network_programming.np2_nonblocking_communication;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Set;

/**
 * 混用阻塞模式与非阻塞模式
 * 把接收客户连接的操作单独由一个线程完成，该线程按照阻塞模式工作
 * 把接收数据和发送数据的操作由另一个线程完成，该线程按照非阻塞模式工作
 * 可提高服务器的并发性能
 * */
public class EchoServerMixedModel {
    private int port = 9099;
    private Selector selector;
    private ServerSocketChannel serverSocketChannel;
    private Charset charset = StandardCharsets.UTF_8;
    private final Object gate = new Object();     // 同步标记，用于多线程同步的对象锁

    public EchoServerMixedModel() throws IOException {
        selector = Selector.open();
        serverSocketChannel = ServerSocketChannel.open();
        serverSocketChannel.socket().setReuseAddress(true);
        serverSocketChannel.socket().bind(new InetSocketAddress(port));
        System.out.println("服务器启动成功");
    }

    public void service() throws IOException {
        while (true) {
            synchronized (gate) {}
            int n = selector.select();
            if (n == 0)
                continue;
            Set<SelectionKey> readyKeys = selector.selectedKeys();
            Iterator<SelectionKey> iterator = readyKeys.iterator();
            while (iterator.hasNext()) {
                SelectionKey key = null;
                try {
                    key = iterator.next();
                    iterator.remove();
                    if (key.isReadable())
                        receive(key);
                    if (key.isWritable())
                        send(key);
                } catch (IOException e) {
                    e.printStackTrace();
                    if (key != null) {
                        try {
                            key.cancel();
                            key.channel().close();
                        } catch (Exception ex) {ex.printStackTrace();}
                    }
                }
            }
        }
    }

    public void accept() {
        while (true) {
            try {
                SocketChannel socketChannel = serverSocketChannel.accept();
                System.out.println("接收到客户连接，来自：" + socketChannel.socket().getInetAddress()
                        + ":" + socketChannel.socket().getPort());
                socketChannel.configureBlocking(false); // 设为非阻塞模式
                ByteBuffer buffer = ByteBuffer.allocate(1024);
                synchronized (gate) {
                    /*
                    * SelectableChannel的register(Selector, ...)和Selector的select()方法都会操纵
                    * Select对象的共享资源all-key集合，SelectableChannel以及Selector的实现对操纵共享
                    * 资源的代码块进行了同步，从而避免对对共享资源的竞争。
                    * 同步机制使得一个线程执行electableChannel的register(Selector, ...)方法时，不允许
                    * 另一个线程同时执行Selector的select()方法，反之亦然。
                    * */
                    selector.wakeup();  // 唤醒执行Selector的select()方法的线程

                    // 新建一个SelectionKey并将其加入Selector的all-keys集合中
                    socketChannel.register(selector, SelectionKey.OP_READ |
                            SelectionKey.OP_WRITE, buffer);
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 处理读就绪事件
     * @param key	SelectionKey对象
     * @throws IOException
     * */
    public void receive(SelectionKey key) throws IOException {
        ByteBuffer buffer = (ByteBuffer) key.attachment();	// 获取与SelectionKey关联的附件
        SocketChannel socketChannel = (SocketChannel) key.channel();	// 获取与SelectionKey关联的SocketChannel
        ByteBuffer readBuf = ByteBuffer.allocate(32);	// 存放读到的数据
        socketChannel.read(readBuf);
        readBuf.flip();		// 把极限设置为位置，把位置设置为0

        buffer.limit(buffer.capacity());	// 把buffer的极限设为容量
        // 把readBuf中的内容拷贝到buffer中，假定buffer的容量足够大，不会出现缓冲区溢出异常
        buffer.put(readBuf);
    }

    /**
     * 处理写就绪事件
     * @param key	SelectionKey对象
     * @throws IOException
     * */
    public void send(SelectionKey key) throws IOException {
        ByteBuffer buffer = (ByteBuffer) key.attachment();	// 获取与SelectionKey关联的附件
        SocketChannel socketChannel = (SocketChannel) key.channel();	// 获取与SelectionKey关联的SocketChannel
        buffer.flip();	// 把极限设置为位置，把位置设置为0
        String data = decode(buffer);	// 按GBK编码把buffer总的字节转换为字符串
        if (data.indexOf("\r\n") == -1)		// 未读到一行数据则返回
            return;
        String outputData = data.substring(0, data.indexOf("\n") + 1);
        System.out.print(outputData);
        ByteBuffer outputBuffer = encode("echo: " + outputData);	// 按GBK编码把输出字符串转换为字节
        // 在非阻塞模式下SocketChannel.write(outputBuffer)方法并不能保证一次就把outputBuffer中的所有字节都发送完，
        // 奉行能发送多少就发送多少的原则，需用循环来保证全部发送
        while (outputBuffer.hasRemaining())		// 判断outputBuffer是否还有剩余
            socketChannel.write(outputBuffer);

        ByteBuffer temp = encode(outputData);
        buffer.position(temp.limit());	// 把buffer的位置设为temp的极限
        buffer.compact();		// 删除缓冲区内从0到当前位置position的内容，即删除已处理的数据

        // 如果已经出现了字符串“bye\r\n”，就使SelectionKey失效，并关闭SocketCannel
        if ("bye\r\n".equals(outputData)) {
            key.cancel();
            socketChannel.close();
            System.out.println("关闭与客户的连接");
        }
    }

    /**
     * 编码
     * */
    private ByteBuffer encode(String str) {
        return charset.encode(str);
    }

    /**
     * 解码
     * */
    private String decode(ByteBuffer buffer) {
        CharBuffer charBuffer = charset.decode(buffer);
        return charBuffer.toString();
    }

    public static void main(String[] args) throws IOException {
        final EchoServerMixedModel server = new EchoServerMixedModel();
        Thread accept = new Thread() {
            public void run() {
                server.accept();
            }
        };
        accept.start();
        server.service();
    }
}
